package com.fyt.reachgo.consummer;

import com.fyt.reachgo.config.RabbitMQConfig;
import com.fyt.reachgo.domain.UserDisinterest;
import com.fyt.reachgo.service.UserDisinterestService;
import com.rabbitmq.client.Channel;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
@RequiredArgsConstructor
public class UserDisinterestConsumer {
    private final UserDisinterestService userDisinterestService;
    // 创建线程缓存区
    private static final int DEFAULT_BATCH_SIZE = 10;
    // 创建线程安全的缓存列表
    private final List<UserDisinterest> batchBuffer = Collections.synchronizedList(new ArrayList<>());
    // 创建一个单线程的调度线程池
    private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

    @PostConstruct
    public void init() {
        // 定时刷新缓冲区(每20秒检查一次)
        executorService.scheduleAtFixedRate(this::flushBatch, 20, 20, TimeUnit.SECONDS);
    }
    @PreDestroy
    public void destroy() {
        executorService.shutdown();
        // 应用关闭前强制刷新剩余消息
        flushBatch();
    }
    @RabbitListener(
            queues = RabbitMQConfig.USER_DISINTEREST_QUEUE,
            ackMode = "MANUAL",
            concurrency = "5-10" // 点赞操作可以更高并发
    )
    @Transactional(rollbackFor = Exception.class)
    public void handleUserDisinterest(UserDisinterest userDisinterest,
                                      Channel channel,
                                      @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){

        // 安全入队
        synchronized (batchBuffer) {
            batchBuffer.add(userDisinterest);
            log.info("用户不感兴趣消息入队成功，当前缓冲区大小为 {}", batchBuffer.size());
            // 如果缓冲区满则刷新缓冲区
            if (batchBuffer.size() >= DEFAULT_BATCH_SIZE) {
                flushBatch();
            }
        }
        // 操作完成后立刻确认消息，防止消息挤压
        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.error("消息确认失败", e);
        }


    }

    private void flushBatch() {
        List<UserDisinterest> messagesToProcess;

        // 1. 安全获取待处理消息
        synchronized (batchBuffer) {
            if (batchBuffer.isEmpty()) {
                log.debug("【用户不感兴趣队列】 ；批量刷新触发，但缓冲区为空，跳过处理");
                return;
            }
            // 获取缓冲区信息
            messagesToProcess = new ArrayList<>(batchBuffer);
            batchBuffer.clear();
        }

        // 2. 空列表二次检查
        if (messagesToProcess.isEmpty()) {
            log.warn("获取到空消息列表，可能并发环境下的竞态条件，跳过处理");
            return;
        }
        log.info("开始批量处理 {} 条用户不感兴趣消息", messagesToProcess.size());
        try {
            userDisinterestService.saveBatch(messagesToProcess);
            log.info("成功批量处理 {} 条用户不感兴趣消息", messagesToProcess.size());
        }catch (Exception e){
            log.error("批量处理用户不感兴趣消息失败", e);

        }
    }

}
